package TGQ.Service

import bean.LanguageSkillMatch
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.sql.functions._
import util.SparkUtil

object LanguageSkillAnalysisService {
  def analyzeLanguageImpact(languageDataStream: DStream[LanguageSkillMatch]): Unit = {
    val spark = SparkUtil.takeSpark()
    import spark.implicits._

    languageDataStream.foreachRDD { rdd =>
      if (!rdd.isEmpty()) {
        val df = rdd.toDF()

        // 炸裂处理多语言（逗号分隔的编程语言列表）
        val explodedDF = df.withColumn("language", explode(split(col("programming_languages"), ",")))

        // 按编程语言和筛选结果分组统计
        val resultDF = explodedDF.groupBy("language", "screening_result")
          .agg(
            count("*").as("数量"),
            round(count(when(col("screening_result") === "通过", 1)) / count("*"), 2).as("通过率")
          )
          .orderBy(desc("通过率"), desc("数量"))

        // 输出分析结果
        println("编程语言对筛选结果影响分析：")
        resultDF.show()

        // 可扩展：保存结果到数据库
        // EducationLevelAnalysisDAO.saveAnalysisResult(resultDF, "language_skill_result")
      }
    }
  }
}